package com.fsql.controller;

import com.dtstack.flink.sql.launcher.utils.JobGraphBuildUtil;
import com.fsql.entity.SqlJobParameterInfo;
import com.fsql.utils.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.io.File;
import java.util.*;

@RestController
@RequestMapping("/jobs")
public class JobController {

    private static Logger LOG = LoggerFactory.getLogger(JobController.class);

    @Autowired
    private Environment env;

    /**
     * 提交sql任务
     * @param jobInfo
     * @return
     */
    @PostMapping("/submitsqljob")
    public ResponseEntity<Map<String, Object>> submitsqljob(@RequestBody SqlJobParameterInfo jobInfo) {
        Map<String, Object> map = new HashMap<>();

        String sqlContent = jobInfo.getSql();

        String sqltempfile = env.getProperty("sqlfile.tempfile");
        if (StringUtils.isBlank(sqltempfile)) {
            sqltempfile = System.getProperty("user.dir");
        }
        String sqlName = sqltempfile + File.separator + UUID.randomUUID().toString() + ".sql";

        FileUtils.writeToFile(sqlName, sqlContent);
        Map<String, Object> conf = new HashMap<>();

        // 任务配置参数
        conf.put("-sql", sqlName);
        LOG.info("sql = {}", sqlName);
        conf.put("-mode", jobInfo.getMode());
        conf.put("-name", jobInfo.getName());
        conf.put("-confProp", jobInfo.getConfProp());
        LOG.info("confProp = {}", jobInfo.getConfProp());

        // TODO 添加jar包得处理
        if (StringUtils.isNotBlank(jobInfo.getAddjars())) {
            conf.put("-addjar", jobInfo.getAddjars());
        }

        conf.put("-localSqlPluginPath", env.getProperty("flinksql.localSqlPluginPath"));
        conf.put("-remoteSqlPluginPath", env.getProperty("flinksql.remoteSqlPluginPath"));
        conf.put("-flinkconf", env.getProperty("flinksql.flinkconf"));
        LOG.info("localSqlPluginPath = {}", env.getProperty("flinksql.localSqlPluginPath"));
        conf.put("-pluginLoadMode", "classpath");
//        conf.put("-checkResource", "true");

        map.put("jobId", JobUtils.submitStandaloneJob(conf));
        return new ResponseEntity<>(map, HttpStatus.OK);
    }

    /**
     * 取消job
     * @param jobId
     * @return
     */
    @GetMapping("/cancel/{jobId}")
    public ResponseEntity<Map<String, Object>> cancelJob(@PathVariable String jobId) {
        Map<String, Object> map = new HashMap<>();
        FlinkRestClientUtils.cancelJob(jobId);
        map.put("result", "取消成功");
        return new ResponseEntity<>(map, HttpStatus.OK);
    }

    /**
     * 获取job的状态
     * @param jobId
     * @return
     */
    @GetMapping("/jobstatus/{jobId}")
    public ResponseEntity<JobStatus> getJobStatus(@PathVariable String jobId) {
        Map<String, Object> map = new HashMap<>();
        return new ResponseEntity<>(FlinkRestClientUtils.getJobStatus(jobId), HttpStatus.OK);
    }

    /**
     * 获取job详细信息
     * @param jobId
     * @return
     */
    @GetMapping("/jobdetail/{jobId}")
    public ResponseEntity<JobDetailsInfo> getJobDetails(@PathVariable String jobId) {
        Map<String, Object> map = new HashMap<>();
        return new ResponseEntity<>(FlinkRestClientUtils.getJobDetails(jobId), HttpStatus.OK);
    }

    /**
     * 获取所有job的状态信息
     * @return
     */
    @GetMapping("/listJobs")
    public ResponseEntity<Collection<JobStatusMessage>> listJobs() {
        Map<String, Object> map = new HashMap<>();
        return new ResponseEntity<>(FlinkRestClientUtils.listJobs(), HttpStatus.OK);
    }

    /**
     * 获取任务的checkpoint配置
     * @param jobId
     * @return
     */
    @GetMapping("/{jobId}/checkpoints/config")
    public ResponseEntity<String> checkpointsConfig(@PathVariable String jobId) {
        String url = String.format("%s/jobs/%s/checkpoints/config", FlinkRestClientUtils.getInterfaceUrl(), jobId);
        return new ResponseEntity<>(HttpUtils.doGet(url), HttpStatus.OK);
    }

    @GetMapping("/{jobId}/checkpoints")
    public ResponseEntity<String> checkpointsStats(@PathVariable String jobId) {
        String url = String.format("%s/jobs/%s/checkpoints", FlinkRestClientUtils.getInterfaceUrl(), jobId);
        return new ResponseEntity<>(HttpUtils.doGet(url), HttpStatus.OK);
    }

    /**
     * 获取任务的最新exception
     * @param jobId
     * @return
     */
    @GetMapping("/{jobId}/exceptions")
    public ResponseEntity<String> jobException(@PathVariable String jobId) {
        String url = String.format("%s/jobs/%s/exceptions", FlinkRestClientUtils.getInterfaceUrl(), jobId);
        return new ResponseEntity<>(HttpUtils.doGet(url), HttpStatus.OK);
    }

    /**
     * 取消任务并返回savePointPath
     * @param params
     * @return
     */
    @PostMapping("/cancelWithSavepoint")
    public ResponseEntity<String> cancelWithSavepoint(Map<String, Object> params) {
        String jobId = String.valueOf(params.get("jobId"));
        if (StringUtils.isBlank(jobId)) {
            return new ResponseEntity<>("任务ID不能为空", HttpStatus.OK);
        }
        String savepointDirectory = String.valueOf(params.get("savepointDirectory"));
        if (StringUtils.isBlank(savepointDirectory)) {
            // TODO 取flink-conf.yaml中的配置
            Configuration flinkConfiguration = JobGraphBuildUtil.getFlinkConfiguration(env.getProperty("flinksql.flinkconf"), new Properties());
            savepointDirectory = flinkConfiguration.getString(ConfigConstant.FLINK_STATE_SAVEPOINTS_DIR);
            if (StringUtils.isBlank(savepointDirectory)) {
                return new ResponseEntity<>("savepointDirectory不能为空", HttpStatus.OK);
            }
        }
        String url = FlinkRestClientUtils.cancelWithSavepoint(jobId, savepointDirectory);
        return new ResponseEntity<>(HttpUtils.doGet(url), HttpStatus.OK);
    }

    /**
     * 停止任务并返回checkpoint路径
     * @param params
     * @return
     */
    @PostMapping("/stopWithSavepoint")
    public ResponseEntity<String> stopWithSavepoint(Map<String, Object> params) {
        String jobId = String.valueOf(params.get("jobId"));
        if (StringUtils.isBlank(jobId)) {
            return new ResponseEntity<>("任务ID不能为空", HttpStatus.OK);
        }
        String savepointDirectory = String.valueOf(params.get("savepointDirectory"));
        if (StringUtils.isBlank(savepointDirectory)) {
            // TODO 取flink-conf.yaml中的配置
            Configuration flinkConfiguration = JobGraphBuildUtil.getFlinkConfiguration(env.getProperty("flinksql.flinkconf"), new Properties());
            savepointDirectory = flinkConfiguration.getString(ConfigConstant.FLINK_STATE_SAVEPOINTS_DIR);
            if (StringUtils.isBlank(savepointDirectory)) {
                return new ResponseEntity<>("savepointDirectory不能为空", HttpStatus.OK);
            }
        }
        String url = FlinkRestClientUtils.stopWithSavepoint(jobId, false, savepointDirectory);
        return new ResponseEntity<>(HttpUtils.doGet(url), HttpStatus.OK);
    }
}
